package com.atbeijing.D02;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;

/**
 * 报警的 Application 接收两条流，一条是温度传感器传回来的数据，一条是烟雾传感器传回来的数据。
 * 当两条流都超过各自的阈值时，报警。
 */
public class Example8 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //温感器流
        DataStreamSource<SensorReading> sensorStream = env.addSource(new SensorSource());
        //烟感器流
        DataStreamSource<SmokeLevel> smokeSource = env.addSource(new SmokeLevelSource());

        //根据温感器id分区,每个温感器有一分区
        KeyedStream<SensorReading, String> sensorKeyedStream = sensorStream.keyBy(new KeySelector<SensorReading, String>() {
            @Override
            public String getKey(SensorReading value) throws Exception {
                return value.id;
            }
        });

        //每个温感器的流连接一个烟感器的流,相当于将烟感器流广播到每个温感器流
        ConnectedStreams<SensorReading, SmokeLevel> connectedStreams = sensorKeyedStream.connect(smokeSource);

        //SensorReading为第一条流类型,SmokeLevel为第二条流类型,Alert为输出流的类型
        SingleOutputStreamOperator<Alert> singStream = connectedStreams.flatMap(new CoFlatMapFunction<SensorReading, SmokeLevel, Alert>() {
            //设置烟感器初始化为LOW,即不报警
            private SmokeLevel smokeLevel = SmokeLevel.LOW;

            //处理温感器的流,当烟感this.smokeLevel为HIGH且温感大80时报警
            //处理完数据不缓存
            @Override
            public void flatMap1(SensorReading sensorReading, Collector<Alert> collector) throws Exception {
                if (smokeLevel == SmokeLevel.HIGH && sensorReading.temperature > 80) {
                    collector.collect(new Alert("着火啦!!!!!", sensorReading.timestamp));
                }
            }

            //处理烟感器的流,根据流传来的数据改变smokeLevel的值
            @Override
            public void flatMap2(SmokeLevel smokeLevel, Collector<Alert> collector) throws Exception {
                this.smokeLevel = smokeLevel;
            }
        });

        //流打印
        singStream.print();

        env.execute();
    }
}
